package com.tzg157.fitness.service;

import cn.hutool.json.JSONObject;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.sql.*;
import org.apache.spark.sql.types.StructType;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.springframework.stereotype.Service;

import java.io.Serializable;
import java.util.Map;

@Service
public class RegisterLog2HiveServiceImpl implements Serializable {

    private static final StructType registerLogSchema = new StructType()
            .add("username", "string")
            .add("gender", "string")
            .add("age","int")
            .add("timestamp_","string");

    public void registerLog2Hive(JavaDStream<Row> rowStream, SparkSession spark) {
        // 将 DStream 转换为 Dataset<Row>
        rowStream.foreachRDD(rdd -> {
            if (!rdd.isEmpty()) {
                Dataset<Row> newData = spark.createDataFrame(rdd, registerLogSchema);
                System.out.println("==========>" + newData.rdd().first().getString(0));
                System.out.println("======================================================");
                newData.write().format("hive").mode(SaveMode.Append).saveAsTable("fitness.ods_member_info");
                System.out.println("=======================================================");
            }
        });
    }

    public void registerLog2Hive(JavaRDD<Row> rowRdd,SparkSession spark){
        Dataset<Row> newData = spark.createDataFrame(rowRdd, registerLogSchema);
        System.out.println("==========>" + newData.rdd().first().getString(0));
        System.out.println("======================================================");
        newData.write().format("hive").mode(SaveMode.Append).saveAsTable("fitness.ods_member_info");
        System.out.println("=======================================================");
    }

//    public void save(JSONObject jsonObject,SparkSession spark){
//        spark.sql("")
//    }

    public Row createRow(JSONObject jsonObject) {
        String type = (String) jsonObject.get("type");
        String username = (String) ((Map)jsonObject.get("data")).get("username");
        String gender = (String) ((Map)jsonObject.get("data")).get("gender");
        Integer age = (Integer) ((Map)jsonObject.get("data")).get("age");
        String timestamp_ = (String) jsonObject.get("timestamp_");
        return RowFactory.create(type, username, gender, age, timestamp_);
    }
}
